package detector

import (
    "sync"
    "time"
    "../common"
)

type FailureDetector struct {
    detections map[common.Node]*Detection
    records    map[common.Node]*Record
    mu         sync.RWMutex
    opt        Option
    myNode     common.Node
    ackChan    chan BeaconAckWithError
    stopping   chan interface{}
    stopped    chan interface{}
    //callbacks
    onConnectedCallback    func(common.Node)
    onDisconnectedCallback func([]common.Node)
}

func (fdc *FailureDetector) CheckAll() {
    fdc.mu.Lock()
    defer fdc.mu.Unlock()
    var expiredNodes []common.Node
    for node, rcd := range fdc.records {
        if _, ok := fdc.detections[node]; !ok {
            delete(fdc.records, node)
            continue
        }
        if rcd.isAlive && common.NowMillisecond()+fdc.opt.CheckIntervalMilliseconds-rcd.lastTime > fdc.opt.LeaseMilliseconds {
            rcd.isAlive = false
            expiredNodes = append(expiredNodes, node)
        }
    }
    if len(expiredNodes) > 0 {
        fdc.onDisconnectedCallback(expiredNodes)
    }
}

func (fdc *FailureDetector) onBeaconAck(ack BeaconAckWithError) {
    if ack.err != nil {
        return
    }
    node := common.Node(ack.ack.PrimeNode)
    fdc.mu.Lock()
    defer fdc.mu.Unlock()
    rcd, ok := fdc.records[node]
    if !ok {
        fdc.records[node] = NewRecord()
        fdc.onConnectedCallback(node)
        return
    }
    if rcd.isAlive == false && common.NowMillisecond()-rcd.lastTime <= fdc.opt.LeaseMilliseconds {
        rcd.isAlive = true
        fdc.onConnectedCallback(node)
    }
}

func (fdc *FailureDetector) IsConnected(node common.Node) bool {
    fdc.mu.RLock()
    defer fdc.mu.RUnlock()
    for n, rcd := range fdc.records {
        if n == node && rcd.isAlive {
            return true
        }
    }
    return false
}

func (fdc *FailureDetector) UnRegister(node common.Node) {
    fdc.mu.Lock()
    defer fdc.mu.Unlock()
    if _, ok := fdc.detections[node]; !ok {
        return
    }
    fdc.detections[node].Stop()
    delete(fdc.detections, node)
}

func (fdc *FailureDetector) Register(node common.Node) {
    fdc.mu.Lock()
    defer fdc.mu.Unlock()
    if _, ok := fdc.detections[node]; ok {
        return
    }
    fdc.detections[node] = NewClientTask(fdc.opt.BeaconIntervalMilliseconds, fdc.opt.DelayMilliseconds, node, fdc.myNode, fdc.ackChan)
}

func (fdc *FailureDetector) Stop() {
    close(fdc.stopping)
    <-fdc.stopped
}

func NewFailureDetector(myNode string, opt Option,
    onConnectedCallback func(common.Node), onDisconnectedCallback func([]common.Node)) *FailureDetector {
    fcd := &FailureDetector{
        detections:             make(map[common.Node]*Detection),
        records:                make(map[common.Node]*Record),
        ackChan:                make(chan BeaconAckWithError),
        opt:                    opt,
        myNode:                 common.Node(myNode),
        onConnectedCallback:    onConnectedCallback,
        onDisconnectedCallback: onDisconnectedCallback,
        stopping:               make(chan interface{}),
        stopped:                make(chan interface{}),
    }
    return fcd
}

func (fdc *FailureDetector) Start() {
    tk := time.Tick(time.Duration(fdc.opt.CheckIntervalMilliseconds) * time.Millisecond)
LOOP:
    for {
        select {
        case ack := <-fdc.ackChan:
            fdc.onBeaconAck(ack)
        case <-tk:
            fdc.CheckAll()
        case <-fdc.stopping:
            fdc.mu.Lock()
            for _, t := range fdc.detections {
                t.Stop()
            }
            fdc.mu.Unlock()
            break LOOP
        }
    }
    close(fdc.stopped)
}
